package com.atguigu.day07;

import com.atguigu.utils.PayEvent;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

// 实时对账
public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<PayEvent> appStream = env
                .addSource(new SourceFunction<PayEvent>() {
                    @Override
                    public void run(SourceContext<PayEvent> ctx) throws Exception {
                        PayEvent e1 = new PayEvent("order-1", "app", 1000L);
                        ctx.collectWithTimestamp(e1, e1.ts);
                        Thread.sleep(1000L);
                        ctx.emitWatermark(new Watermark(999L));
                        Thread.sleep(1000L);
                        PayEvent e2 = new PayEvent("order-2", "app", 2000L);
                        ctx.collectWithTimestamp(e2, e2.ts);
//                        Thread.sleep(1000L);
                        ctx.emitWatermark(new Watermark(7000L));
                        Thread.sleep(1000L);
                    }

                    @Override
                    public void cancel() {

                    }
                });

        DataStreamSource<PayEvent> payStream = env
                .addSource(new SourceFunction<PayEvent>() {
                    @Override
                    public void run(SourceContext<PayEvent> ctx) throws Exception {
                        PayEvent e1 = new PayEvent("order-1", "weixin", 4000L);
                        ctx.collectWithTimestamp(e1, e1.ts);
                        Thread.sleep(1000L);
                        ctx.emitWatermark(new Watermark(7000L));
                        Thread.sleep(2000L);
                        PayEvent e2 = new PayEvent("order-2", "weixin", 9000L);
                        ctx.collectWithTimestamp(e2, e2.ts);
                        Thread.sleep(1000L);
                    }

                    @Override
                    public void cancel() {

                    }
                });

        appStream.keyBy(r -> r.orderId)
                .connect(payStream.keyBy(r -> r.orderId))
                .process(new Match())
                .print();

        env.execute();
    }

    public static class Match extends CoProcessFunction<PayEvent, PayEvent, String> {
        //如果app支付先到达就保存在state1里
        private ValueState<PayEvent> state1;
        //如果微信支付先到达就保存在state2里
        private ValueState<PayEvent> state2;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            state1 = getRuntimeContext().getState(
                    new ValueStateDescriptor<PayEvent>("state1", Types.POJO(PayEvent.class))
            );
            state2 = getRuntimeContext().getState(
                    new ValueStateDescriptor<PayEvent>("state2", Types.POJO(PayEvent.class))
            );
        }

        @Override
        public void processElement1(PayEvent value, Context ctx, Collector<String> out) throws Exception {
            if (state2.value() != null) {
                // 如果state2不为空，说明weixin支付比app支付先到达
                out.collect("订单：" + value.orderId + "对账成功，weixin支付先到达");
                // 清空保存weixin支付的状态变量state2
                state2.clear();
            } else {
                // 说明app支付先到达，将app支付保存到state1中
                state1.update(value);
                // 注册一个5秒钟之后的定时器
                ctx.timerService().registerEventTimeTimer(value.ts + 5000L);
            }
        }

        @Override
        public void processElement2(PayEvent value, Context ctx, Collector<String> out) throws Exception {
            if (state1.value() != null) {
                // 如果state1不为空，说明app支付比weixin支付先到达
                out.collect("订单：" + value.orderId + "对账成功，app支付先到达");
                state1.clear();
            } else {
                // 说明weixin支付先到达，将weixin支付保存到state2中
                state2.update(value);
                // 注册一个5秒钟之后的定时器
                ctx.timerService().registerEventTimeTimer(value.ts + 5000L);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            if (state1.value() != null) {
                out.collect("订单" + state1.value().orderId + "对账失败，对应的weixin支付没来");
                state1.clear();
            }
            if (state2.value() != null) {
                out.collect("订单" + state2.value().orderId + "对账失败，对应的app支付没来");
                state2.clear();
            }
        }
    }
}
